iT邦幫忙

2024 iThome 鐵人賽

DAY 11
0
Python

時空序列分析-關鍵籌碼分析系列 第 11

python繪製股票K線圖第二彈! 上櫃與上市公司的股價爬蟲

  • 分享至 

  • xImage
  •  

基於昨天可以把上市公司的股價爬蟲寫出來,
今天把上櫃公司也做一遍~

延續昨天的那份程式,上面這邊套件引用、圖表設定的,影響不大的都保留著

import pandas as pd
import requests
import re
import time
import mplfinance as mpf
import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
from datetime import datetime
# 設定字體
zhfont = fm.FontProperties(fname=r'C:\Windows\Fonts\kaiu.ttf')  # 使用原始字串
plt.rcParams['font.family'] = zhfont.get_name()

台股上櫃公司-抓取網址(櫃台買賣中心資料)

http://www.tpex.org.tw/web/stock/aftertrading/daily_trading_info/st43_result.php?d=112/08&stkno=6223

測試一下

json_data=requests.get('http://www.tpex.org.tw/web/stock/aftertrading/daily_trading_info/st43_result.php?d=112/08&stkno=6223').json()

columns= ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']

#只取json格式資料的aaData內的資料哦!
data=pd.DataFrame(json_data['aaData'],columns=columns)
data.head()

日期 成交股數 成交金額 開盤價 最高價 最低價 收盤價 漲跌價差 成交筆數
0 112/08/01 5,583 1,143,509 207.00 213.00 199.00 203.00 -2.00 4,562
1 112/08/02 6,643 1,276,902 203.00 203.50 185.50 187.50 -15.50 5,802
2 112/08/04 3,463 659,699 190.00 194.00 187.00 190.50 3.00 3,020
3 112/08/07 2,752 529,006 190.50 195.50 187.00 194.00 3.50 2,483
4 112/08/08 2,140 409,123 194.00 194.50 189.00 191.00 -3.00 2,326

OK~ 可以正常運作! YES!

重新觀察過參考的這篇文章後,

作者將"市場別"欄位也爬取下來用來辨識是"上市"還是"上櫃"

只不過我沒用到那個欄位,就寫一個陽春版的手動調整型工人智慧 :D

(其實主要是懶癌發作(? )

定義一個根據上面下載"上市公司"資料的function

# 定義下載資料的函數
def fetch_data(date, stock_no):
    url = f'http://www.twse.com.tw/exchangeReport/STOCK_DAY?response=json&date={date}&stockNo={stock_no}'
    max_retries = 5
    for i in range(max_retries):
        try:
            response = requests.get(url)
            response.raise_for_status()  # 如果請求失敗,則引發異常
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f'Error fetching data for {date}: {e}. Retry {i + 1}/{max_retries}')
            time.sleep(3)
    raise Exception(f'Failed to fetch data for {date} after {max_retries} retries')

模仿上面的寫法稍微修改一下,

定義一個根據上面下載"上櫃公司"資料的function

def fetch_tpex_data(date, stock_no):
    url = f'https://www.tpex.org.tw/web/stock/aftertrading/daily_trading_info/st43_result.php?d={date}&stkno={stock_no}'
    max_retries = 5
    for i in range(max_retries):
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f'Error fetching TPEX data for {date}: {e}. Retry {i + 1}/{max_retries}')
            time.sleep(3)
    raise Exception(f'Failed to fetch TPEX data for {date} after {max_retries} retries')

昨天下載資料的function:

# 定義下載和合併資料的函數
def download_stock_data(start_year, start_month, end_year, end_month, stock_no='2330'):
    data_frames = []
    for year in range(start_year, end_year + 1):
        start_m = start_month if year == start_year else 1
        end_m = end_month if year == end_year else 12
        for month in range(start_m, end_m + 1):
            date_str = f'{year}{month:02d}01'
            try:
                json_data = fetch_data(date_str, stock_no)
                columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
                df = pd.DataFrame(json_data['data'], columns=columns)
                data_frames.append(df)
            except Exception as e:
                print(f'Error fetching data for {date_str}: {e}')
            # 跳過未來的月份
            current_date = datetime.now()
            if year == current_date.year and month >= current_date.month:
                break
    all_data = pd.concat(data_frames, ignore_index=True)
    return all_data

那我就在中間抓json資料的地方寫個簡單的if else判斷一下
TWSE就是上市,不是TWSE就是TPEX

if market == 'TWSE':
json_data = fetch_twse_data(date_str, stock_no)
columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
df = pd.DataFrame(json_data['data'], columns=columns)
else: # TPEX
json_data = fetch_tpex_data(date_str, stock_no)
columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
df = pd.DataFrame(json_data['aaData'], columns=columns)
data_frames.append(df)

完整寫起來就長這樣
直接把stock_no(股票代號)、market(市場別)的參數設定都放到主程序那,可自由調整

def download_stock_data(start_year, start_month, end_year, end_month, stock_no, market):
    data_frames = []
    for year in range(start_year, end_year + 1):
        start_m = start_month if year == start_year else 1
        end_m = end_month if year == end_year else 12
        for month in range(start_m, end_m + 1):
            date_str = f'{year}{month:02d}01' if market == 'TWSE' else f'{year-1911}/{month:02d}'
            try:
                if market == 'TWSE':
                    json_data = fetch_twse_data(date_str, stock_no)
                    columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
                    df = pd.DataFrame(json_data['data'], columns=columns)
                else:  # TPEX
                    json_data = fetch_tpex_data(date_str, stock_no)
                    columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
                    df = pd.DataFrame(json_data['aaData'], columns=columns)
                data_frames.append(df)
            except Exception as e:
                print(f'Error fetching data for {date_str}: {e}')
            current_date = datetime.now()
            if year == current_date.year and month >= current_date.month:
                break
    all_data = pd.concat(data_frames, ignore_index=True)
    return all_data

因為TWSETPEX提供的格式有點不太一樣,
原本照著寫會出錯,debug了一下,來為上櫃特別寫

if market == 'TPEX':
        # 民國年轉換為公元年
        def convert_to_gregorian(date_str):
            parts = date_str.split('/')
            parts[0] = str(int(parts[0]) + 1911)
            return '/'.join(parts)
        all_data['日期'] = all_data['日期'].apply(convert_to_gregorian)

完整寫起來變成這樣

def clean_data(all_data, market):
    if market == 'TPEX':
        # 民國年轉換為公元年
        def convert_to_gregorian(date_str):
            parts = date_str.split('/')
            parts[0] = str(int(parts[0]) + 1911)
            return '/'.join(parts)

        all_data['日期'] = all_data['日期'].apply(convert_to_gregorian)

    all_data[['成交股數', '成交金額', '成交筆數']] = all_data[['成交股數', '成交金額', '成交筆數']].replace(',', '', regex=True)
    
    def to_float(x):
        try:
            return float(x)
        except ValueError:
            return float('nan')

    for col in all_data.columns[1:]:
        all_data[col] = all_data[col].map(to_float)
    
    all_data[['成交股數', '成交金額']] = all_data[['成交股數', '成交金額']] / 1000
    all_data = all_data.rename(columns={'成交股數': '成交張數'})
    all_data = all_data.dropna()

    # 轉換日期格式
    try:
        all_data['日期'] = pd.to_datetime(all_data['日期'], format='%Y/%m/%d')
    except ValueError:
        all_data['日期'] = pd.to_datetime(all_data['日期'], format='%Y/%m/%d', errors='coerce')
    all_data = all_data.dropna(subset=['日期'])

    all_data = all_data.rename(columns={
        '開盤價': 'Open',
        '最高價': 'High',
        '最低價': 'Low',
        '收盤價': 'Close',
        '成交張數': 'Volume'
    })

    all_data['Open'] = all_data['Open'].astype(float)
    all_data['High'] = all_data['High'].astype(float)
    all_data['Low'] = all_data['Low'].astype(float)
    all_data['Close'] = all_data['Close'].astype(float)
    all_data['Volume'] = all_data['Volume'].astype(float)

    all_data.set_index('日期', inplace=True)
    return all_data

後面主程序的部分就可以自由調整想要的日期區間股票代號市場別

(我這邊直接故意把時間設定到今年12月,它只會抓到現有的所有最新資料)

# 主程序
start_year = 2019
start_month = 1
end_year = 2024
end_month = 12
stock_no = '6223'
market = 'TPEX'

#呼叫函式
all_data = download_stock_data(start_year, start_month, end_year, end_month, stock_no, market)
cleaned_data = clean_data(all_data, market)

繪圖記得給他畫出來

import mplfinance as mpf

# 繪製 K 線圖
def plot_data(all_data, title):
    fig, ax = plt.subplots()

    # 自定義樣式
    custom_style = mpf.make_mpf_style(
        base_mpf_style='charles',
        marketcolors={
            'candle': {'up': 'red', 'down': 'green'},
            'edge': {'up': 'red', 'down': 'green'},
            'wick': {'up': 'red', 'down': 'green'},
            'ohlc': {'up': 'red', 'down': 'green'},
            'volume': {'up': 'red', 'down': 'green'},
            'alpha': 1.0
        }
    )

    # 繪圖
    mpf.plot(all_data, type='candle', style=custom_style, ax=ax, warn_too_much_data=len(all_data) + 1)
    ax.set_title(title, fontproperties=zhfont)
    ax.set_ylabel('價格', fontproperties=zhfont)
    plt.show()

最後plot出來

plot_data(cleaned_data, f'旺矽 {start_year}/{start_month} - {end_year}/{end_month} 每日股票交易價格和收盤情況')

依自己想要的時間區間,這樣手刻的股價K線圖就出來了~

這邊以上櫃公司"旺矽(6223)"來舉例(如下圖)。

https://ithelp.ithome.com.tw/upload/images/20240810/20168322YHNFrSpBX8.png


每日記錄:
明天又要開盤了,看戲 (吃爆米花
其實K線圖畫出來,如果自己想要進行回測或是寫策略就可以在後面加上去~
或是其實用其他人家軟體、平台提供寫好的工具進行回測和寫策略也可以,
個人習慣問題,自己寫比較苦,但是很自由。

然後變成程式交易(誤

之後要是後面做完還沒滿三十天,來放一些debug的過程好了(陷入重重危機)。

恭喜林郁婷為她也為台灣奪下史上女子拳擊第一面奧運金牌!!
當下看到那一刻只有滿滿的感動,瞬間熱淚盈眶,
小時候我也被霸凌過,要怎麼想像一個女生前面27年來是怎麼承受這些耳語的,
她代表著台灣,用善良,來對抗世界的霸凌。 🥹


上一篇
爬蟲拿資料? 來學習怎麼畫股價K線圖! Step(2/2): 自己爬股價
下一篇
爬蟲很好玩,獲取籌碼資訊有解? 動態網頁的爬蟲方式!
系列文
時空序列分析-關鍵籌碼分析31
圖片
  直播研討會
圖片
{{ item.channelVendor }} {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言